package org.example.flink.operation;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class Hello12SinkToKafka {
    public static void main(String[] args) throws Exception {
        //运行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(2);
        //操作数据
        DataStreamSource<String> source = environment.socketTextStream("localhost", 19523);

        //Kafka输出配置
        KafkaSink<String> kafkaSinkSetting = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.101.100:9092,192.168.101.101:9092,192.168.101.102:9092")
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic("flink_sink")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        source.map(word -> word + ":" + System.currentTimeMillis()).sinkTo(kafkaSinkSetting);

        //运行环境
        environment.execute();
    }
}
